package com.hitqz.robot.dispatch.netty.listener;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.io.FileUtil;
import cn.hutool.core.io.IORuntimeException;
import cn.hutool.core.util.StrUtil;
import cn.hutool.core.util.URLUtil;
import cn.hutool.core.util.ZipUtil;
import cn.hutool.extra.spring.SpringUtil;
import com.alibaba.excel.EasyExcel;
import com.alibaba.excel.write.style.column.LongestMatchColumnWidthStyleStrategy;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.nacos.shaded.com.google.common.util.concurrent.RateLimiter;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ArrayNode;
import com.google.common.cache.Cache;
import com.hitqz.robot.api.business.dto.*;
import com.hitqz.robot.api.business.entity.*;
import com.hitqz.robot.api.business.params.QueryPatrolSummaryParams;
import com.hitqz.robot.api.business.util.ExcelUtil;
import com.hitqz.robot.api.business.vo.PatrolResultSummaryVo;
import com.hitqz.robot.api.dispatch.model.MessageDto;
import com.hitqz.robot.api.dispatch.model.RobotDispatchDto;
import com.hitqz.robot.api.dispatch.protocol.ProtocolEnum;
import com.hitqz.robot.api.dispatch.protocol.dto.*;
import com.hitqz.robot.api.dispatch.protocol.dto.RobotWarningDto;
import com.hitqz.robot.biz.config.MultiColumnMergeHandler;
import com.hitqz.robot.biz.service.*;
import com.hitqz.robot.biz.websocket.endpoint.SubscriptionEndpoint;
import com.hitqz.robot.biz.websocket.endpoint.SubscriptionLogEndpoint;
import com.hitqz.robot.common.core.jackson.RobotJavaTimeModule;
import com.hitqz.robot.common.core.util.R;
import com.hitqz.robot.common.core.util.RedisUtils;
import com.hitqz.robot.common.file.core.FileProperties;
import com.hitqz.robot.dispatch.netty.ChannelCache;
import com.hitqz.robot.dispatch.netty.factory.AlgorithmFactory;
import com.hitqz.robot.dispatch.netty.util.JtsUtil;
import com.hitqz.robot.dispatch.netty.util.LockUtil;
import com.hitqz.robot.dispatch.netty.util.RobotReplyUtil;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.locationtech.jts.geom.Polygon;
import org.springframework.beans.BeanUtils;
import org.springframework.context.event.EventListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.io.File;
import java.io.IOException;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.net.URL;
import java.nio.charset.StandardCharsets;
import java.nio.file.Paths;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.locks.Lock;
import java.util.function.Function;
import java.util.stream.Collectors;

@Component
@Slf4j
@AllArgsConstructor
public class QueryListener {

    private final FileProperties fileProperties;

    private final RobotService robotService;

    private final RobotMapService robotMapService;

    private final RobotMapPosService robotMapPosService;

    private final RobotMapAreaService robotMapAreaService;

    private final RobotMapAreaPosService robotMapAreaPosService;

    private final TicketsService ticketsService;

    private final TicketsTaskService ticketsTaskService;

    private final RobotWarningRecordService robotWarningRecordService;

    private final RobotGroupService robotGroupService;

    private final AlgorithmFactory algorithmFactory;

    private final PatrolResultService patrolResultService;

    private final BigDecimal THOUSAND = new BigDecimal(1000);


    private static final ObjectMapper objectMapper = new ObjectMapper()
            .registerModule(new RobotJavaTimeModule());
    public static final Map<String, Map<String, String>> topicClientMessages = new ConcurrentHashMap<>(16);

    private static final Map<String, String> CLIENT_MAP_SERVER_URL_MAP = new HashMap<>(8);

    //话题限流
    private static final Map<String, RateLimiter> topicRateLimiterMap = new ConcurrentHashMap<>(8);


    @EventListener
    @Async("serverListenerThreadPool")
    public void handleQuery(RobotMapDto dto) {
        if (log.isDebugEnabled()) {
            log.debug("[netty listener]" + JSONObject.toJSONString(dto));
        }
        RobotEntity robotEntity = robotService.selectByRobotSn(dto.getClientId());
        if (robotEntity == null) {
            log.info("[netty listener] robot is not registered");
            return;
        }
        RobotMapEntity robotMapEntity = robotMapService.findByMapCode(dto.getMapCode());
        if (robotMapEntity == null) {
            robotMapEntity = new RobotMapEntity();
            BeanUtils.copyProperties(dto, robotMapEntity);
        }
        robotEntity.setMapCode(dto.getMapCode());
        robotService.updateById(robotEntity);
        if (StrUtil.isNotBlank(dto.getDownloadUrl())) {
            try {
                String suffix = robotEntity.getGroupId() + (Objects.equals(dto.getType(), 2) ? ".pcd" : ".png");
                String filePath = StrUtil.isBlank(robotMapEntity.getFilePath()) ?
                        Paths.get(fileProperties.getLocal().getBasePath(), "map", dto.getMapCode() + suffix).toFile().getAbsolutePath()
                        : Paths.get(fileProperties.getLocal().getBasePath(), robotMapEntity.getFilePath()).toString();
                String relativePath = filePath.replace(Paths.get(fileProperties.getLocal().getBasePath()).toString(), "").replaceAll("\\\\", "/");
                try {
                    log.info("map download url:{},{}", dto.getDownloadUrl(), filePath);
                    URL url = URLUtil.url(dto.getDownloadUrl());
                    FileUtil.writeFromStream(URLUtil.getStream(url), filePath);
                    robotMapEntity.setFilePath(relativePath);
                    log.info("{}，更新地图文件 relativePath:{}", dto.getClientId(), relativePath);
                } catch (Exception e) {
                    log.info("下载地图失败" + e.getMessage(), e);
                }
            } catch (Exception e) {
                log.error("sync map download error", e);
            }
        }
        BeanUtil.copyProperties(dto, robotMapEntity, false);
        robotMapService.saveOrUpdate(robotMapEntity);
        //清除旧缓存
        CLIENT_MAP_SERVER_URL_MAP.remove(Optional.of(dto).map(RobotMapDto::getMapCode).orElse("-"));
        //当前所有在用的地图
        List<RobotEntity> robotList = robotService.list();
        List<String> mapCodeList = robotList.stream().map(RobotEntity::getMapCode).filter(StrUtil::isNotBlank).toList();
        //删除地图
        robotMapService.remove(Wrappers.<RobotMapEntity>lambdaQuery().notIn(RobotMapEntity::getMapCode, mapCodeList));
        //删除点位
        robotMapPosService.remove(Wrappers.<RobotMapPosEntity>lambdaQuery().notIn(RobotMapPosEntity::getMapCode, mapCodeList));
    }


    @EventListener
    @Async("serverListenerThreadPool")
    public void handleQuery(RobotMapPosDto dto) {
        if (log.isDebugEnabled()) {
            log.debug("[netty listener]" + JSONObject.toJSONString(dto));
        }
        synchronized (this) {
            RobotEntity robotEntity = robotService.selectByRobotSn(dto.getClientId());
            if (robotEntity == null) {
                log.info("[netty listener] robot is not registered");
            }
            RobotMapPosEntity robotMapPos = robotMapPosService.getOne(Wrappers.<RobotMapPosEntity>lambdaQuery()
                    .eq(RobotMapPosEntity::getMapCode, dto.getMapCode())
                    .eq(RobotMapPosEntity::getCode, dto.getCode())
            );
            if (robotMapPos == null) {
                robotMapPos = new RobotMapPosEntity();
            }
            BeanUtil.copyProperties(dto, robotMapPos, false);
            robotMapPosService.saveOrUpdate(robotMapPos);
        }
    }

    @EventListener
    @Async("serverListenerThreadPool")
    public void handleQuery(RobotMapAreaDto dto) throws JsonProcessingException {
        log.info("[netty listener RobotMapAreaDto]收到消息");
        if (log.isDebugEnabled()) {
            log.debug("[netty listener]" + objectMapper.writeValueAsString(dto));
        }
        synchronized (this) {
            RobotEntity robotEntity = robotService.selectByRobotSn(dto.getClientId());
            if (robotEntity == null) {
                log.info("[netty listener] robot is not registered");
            } else {
                // 全部删除 然后重新添加
                RobotMapAreaEntity areaEntity = new RobotMapAreaEntity();
                BeanUtil.copyProperties(dto, areaEntity, false);
                robotMapAreaService.save(areaEntity);
                if (CollectionUtil.isNotEmpty(dto.getAreaPos())) {
                    for (RobotMapAreaPosDto robotMapAreaPosDto : dto.getAreaPos()) {
                        RobotMapAreaPosEntity posEntity = new RobotMapAreaPosEntity();
                        BeanUtil.copyProperties(robotMapAreaPosDto, posEntity);
                        posEntity.setAreaId(areaEntity.getId());
                        robotMapAreaPosService.save(posEntity);
                    }
                }
            }
        }

    }


    //高频事件 要避免重复查询，否则可能堆溢出
    @EventListener
    @Async("serverListenerThreadPool")
    public void handleQuery(RobotClientSummaryDto dto) {
        try {
            if (log.isDebugEnabled()) {
                log.debug("[netty listener]" + JSONObject.toJSONString(dto));
            }
            //通过ws给前端广播消息
            // 保存下最新的信息
            Cache<String, RobotClientSummaryDto> posInfoDtoCache = SpringUtil.getBean("robotCache");
            if (Optional.of(dto).map(RobotClientSummaryDto::getPosition).isPresent()) {
                RobotPositionDto position = dto.getPosition();
                String mapCode = position.getMapCode();
                //替换地图 url 为 server 本地的
                if (StrUtil.isNotBlank(mapCode)) {
                    String mapUrl = CLIENT_MAP_SERVER_URL_MAP.get(mapCode);
                    if (mapUrl == null) {
                        RobotMapEntity robotMapEntity = robotMapService.findByMapCode(mapCode);
                        //存一个空避免重复查
                        mapUrl = Optional.ofNullable(robotMapEntity).map(RobotMapEntity::getFilePath).orElse("");
                        CLIENT_MAP_SERVER_URL_MAP.put(mapCode, Optional.ofNullable(robotMapEntity).map(RobotMapEntity::getFilePath).orElse(""));
                    }
                    position.setUrl(mapUrl);
                }
            }
            RobotEntity robotEntity = robotService.selectByRobotSn(dto.getClientId());
            if (robotEntity != null) {
                dto.setGroupId(robotEntity.getGroupId());
            }
            posInfoDtoCache.put(dto.getClientId(), dto);
            this.dealWithWarning(dto);
            String topic = "summary";//topic应该整理成枚举类
            topicClientMessages.computeIfAbsent(topic,
                    k -> new ConcurrentHashMap<>()).put(dto.getClientId(), objectMapper.writeValueAsString(dto));
            broadcastMessageByTopic(topic);

            // 计算位置的区域
            Cache<String, Polygon> robotPosPolygon = SpringUtil.getBean("robotPosPolygon");
            BigDecimal robotWidth = new BigDecimal(dto.getOtherInfo().getRobotWidth())
                    .divide(THOUSAND, 3, RoundingMode.HALF_UP);
            BigDecimal robotLength = new BigDecimal(dto.getOtherInfo().getRobotLength())
                    .divide(THOUSAND, 3, RoundingMode.HALF_UP);
            robotPosPolygon.put(dto.getClientId(),
                    JtsUtil.getRectangleVertices(dto.getPosition().getLaserPos().getX(),
                            dto.getPosition().getLaserPos().getY(),
                            robotLength.doubleValue(), robotWidth.doubleValue()));
        } catch (Exception e) {
            log.error(e.getMessage(), e);
        }
    }

    private void dealWithWarning(RobotClientSummaryDto dto) {
        String robotName = dto.getStatus().getName();
        List<com.hitqz.robot.api.business.dto.RobotWarningDto> warningDtoList = Optional.ofNullable(dto.getWarning())
                .orElse(new ArrayList<>());
        List<RobotWarningRecordEntity> warningList = robotWarningRecordService.list(Wrappers
                .<RobotWarningRecordEntity>lambdaQuery()
                .like(RobotWarningRecordEntity::getTitle, "定时检测:%")
                .orderByDesc(RobotWarningRecordEntity::getTime)
                .last("limit 10")
        );
        for (RobotWarningRecordEntity warningRecordEntity : warningList) {
            com.hitqz.robot.api.business.dto.RobotWarningDto warningDto = new com.hitqz.robot.api.business.dto.RobotWarningDto();
            BeanUtil.copyProperties(warningRecordEntity, warningDto);
            String title = warningRecordEntity.getTitle();
            title = title.replaceAll("定时检测:", "");
            warningDto.setTitle(title);
            warningDto.setContent(warningRecordEntity.getContent());
            warningDto.setTime(DateUtil.format(warningRecordEntity.getTime(), "yyyy-MM-dd HH:mm:ss"));
            warningDtoList.add(warningDto);
        }
        warningDtoList.sort(Comparator.comparing(com.hitqz.robot.api.business.dto.RobotWarningDto::getTime).reversed());
        List<com.hitqz.robot.api.business.dto.RobotWarningDto> firstTenWarnings = warningDtoList
                .subList(0, Math.min(warningDtoList.size(), 10));
        dto.setWarning(firstTenWarnings);
    }

    @EventListener
    @Async("serverListenerThreadPool")
    public void handleQuery(RobotDto dto) {
        if (log.isDebugEnabled()) {
            log.debug("[netty listener]" + JSONObject.toJSONString(dto));
        }
        RobotEntity robotEntity = robotService.selectByRobotSn(dto.getRobotSn());
        if (robotEntity == null) {
            robotEntity = new RobotEntity();
            robotEntity.setGroupId(1);

        }
        BeanUtil.copyProperties(dto, robotEntity, false);
        robotService.saveOrUpdate(robotEntity);
    }

    @EventListener
    @Async("serverListenerThreadPool")
    public void handleQuery(ReportLogMessageDto reportLogMessageDto) {
        SubscriptionLogEndpoint.broadcast(
                reportLogMessageDto.getRandomId() == null ? null : Integer.valueOf(reportLogMessageDto.getRandomId()),
                reportLogMessageDto.getClientId(), reportLogMessageDto.getMessage());
    }

    @EventListener
    @Async("serverListenerThreadPool")
    public void handleQuery(RobotApplyPathDto dto) {
        if (log.isDebugEnabled()) {
            log.debug("[netty listener]" + JSONObject.toJSONString(dto));
        }
        synchronized (this) {
            log.info("请求前往目标点:" + dto.getCode() + "---" + dto.getClientId());
            Cache<String, RobotDispatchDto> posInfoDtoCache = SpringUtil.getBean("cacheRobotTask");
            RobotDispatchDto robotDispatchDto = new RobotDispatchDto();
            robotDispatchDto.setClientId(dto.getClientId());
            robotDispatchDto.setPathCodes(dto.getPathList());
            robotDispatchDto.setTargetPosCode(dto.getCode());
            robotDispatchDto.setPathDistances(dto.getPathDistances());
            posInfoDtoCache.put(dto.getClientId(), robotDispatchDto);
            // 回复接受成功 暂时不需要回复回复直接加入调度组缓存里面进行计算
            // 考虑---先暂停队列，后续根据调度的计算判断是否要进行继续
            RobotSendClientMsgDto robotSendClientMsgDto = new RobotSendClientMsgDto();
            robotSendClientMsgDto.setCmd(ProtocolEnum.RESULT.getCmd());
            robotSendClientMsgDto.setSeq(Integer.parseInt(
                    String.valueOf(ProtocolEnum.ROBOT_APPLY_TARGET_POS.getCmd())));
            robotSendClientMsgDto.setClientId(dto.getClientId());
            robotSendClientMsgDto.setData(R.ok());
            SpringUtil.getApplicationContext().publishEvent(robotSendClientMsgDto);
            // 加入调度的数据
//            RobotEntity robotEntity = robotService.selectByRobotSn(dto.getClientId());
//            if (robotEntity != null) {
//                RobotGroupEntity robotGroup = robotGroupService.getById(robotEntity.getGroupId());
//                if (robotGroup == null || StrUtil.isEmpty(robotGroup.getDispatchAlgorithm())) {
//                    log.info("robot has no DispatchAlgorithm or dispatch group {}", dto.getClientId());
//                    return;
//                }
//                algorithmFactory.getRoutingStrategy(robotGroup.getDispatchAlgorithm()).changePath(robotGroup, dto);
//            }
        }


    }


    @EventListener
    @Async("serverListenerThreadPool")
    public void handlerQuery(FormConfigModalDto dto) {
        //使用redis，而不用cache，避免重启server，而client没有连接，且要去修改旧任务，会没有数据。
        String jsonString = JSONObject.toJSONString(dto);
        RedisUtils.set("dynamicTaskContent:" + dto.getRobotSn(), jsonString);
        log.info("保存动态表单:{},{},{}", dto.getRobotSn(), dto.getFormTitle(), dto.getFormName());
    }


    @EventListener
    @Async("serverListenerThreadPool")
    public void handleQuery(RobotTicketsDto dto) {
        if (log.isDebugEnabled()) {
            log.debug("[netty listener]" + JSONObject.toJSONString(dto));
        }
        synchronized (dto.getAllotRobotSn().intern()) {
            //是否需要同步工单/工单任务在服务端数据库id到执行端
            boolean syncServerIdFlag = false;
            TicketsEntity tickets = new TicketsEntity();
            if (dto.getServerId() != null) {
                tickets = ticketsService.getById(dto.getServerId());
                if (tickets != null && tickets.getStatus() == 4) {
                    log.info("工单:{}已完成，不用更新相关信息", tickets);
                    return;
                }
            } else {
                syncServerIdFlag = true;
            }
            BeanUtil.copyProperties(dto, tickets);
            ticketsService.saveOrUpdate(tickets);
            dto.setServerId(tickets.getId().intValue());
            if (dto.getTickets() != null && !dto.getTickets().isEmpty()) {
                List<TicketsTaskEntity> updateList = new ArrayList<>();
                List<TicketsTaskEntity> saveList = new ArrayList<>();
                for (RobotTicketsTaskDto taskDto : dto.getTickets()) {
                    TicketsTaskEntity ticketsTaskEntity = new TicketsTaskEntity();
                    if (taskDto.getServerId() != null) {
                        ticketsTaskEntity = ticketsTaskService.getById(taskDto.getServerId());
                        updateList.add(ticketsTaskEntity);
                    } else {
                        syncServerIdFlag = true;
                        saveList.add(ticketsTaskEntity);
                    }
                    BeanUtil.copyProperties(taskDto, ticketsTaskEntity, false);
                    ticketsTaskEntity.setTicketsId(tickets.getId());
                }
                if (!updateList.isEmpty()) {
                    ticketsTaskService.updateBatchById(updateList);
                    log.info("批量更新工单任务成功");
                }
                if (!saveList.isEmpty()) {
                    ticketsTaskService.saveBatch(saveList);
                    log.info("批量保存工单任务成功");
                }
                List<TicketsTaskEntity> list = ticketsTaskService.list(Wrappers.<TicketsTaskEntity>lambdaQuery().eq(TicketsTaskEntity::getTicketsId, tickets.getId()));
                //对比 list的content和dto的tickets的content是否相同，相同则赋予serverId
                // 将数据库查询结果映射为 Map<content, entity>
                Map<String, TicketsTaskEntity> entityMap = list.stream()
                        .collect(Collectors.toMap(TicketsTaskEntity::getContent, Function.identity(), (e1, e2) -> e1));
                // 遍历 DTO 列表，根据 content 从 Map 中查找 fixme content不能重复
                for (RobotTicketsTaskDto taskDto : dto.getTickets()) {
                    TicketsTaskEntity matchedEntity = entityMap.get(taskDto.getContent());
                    if (matchedEntity != null) {
                        taskDto.setServerId(matchedEntity.getId().intValue());
                    }
                }

            }
            // 更新完成后，下发serverId
            if (syncServerIdFlag) {
                MessageDto messageDto = new MessageDto();
                messageDto.setClientId(dto.getAllotRobotSn());
                messageDto.setSn(ProtocolEnum.ROBOT_TICKETS.getCmd());
                messageDto.setCmd(ProtocolEnum.ROBOT_TICKETS.getCmd());
                messageDto.setData(JSONObject.toJSONString(dto).getBytes(StandardCharsets.UTF_8));
                ChannelCache.send2Client(dto.getAllotRobotSn(), messageDto);
            }

            //这里是耗时操作,尽量少执行; 暂停、停止、完成、失败才生成
            List<Integer> statusList = Arrays.asList(2, 3, 4, 5);
            if (statusList.contains(dto.getStatus())) {
                // 任务完成后，生成巡检报表excel
                log.info("{},tickets server id:{},生成excel...", dto.getAllotRobotSn(), dto.getServerId());
                Integer ticketsId = dto.getServerId();
                String patrolExcelFilePath = ExcelUtil.getPatrolExcelFilePath(fileProperties.getLocal().getBasePath(), ticketsId);
                QueryPatrolSummaryParams query = new QueryPatrolSummaryParams();
                query.setTicketsId(ticketsId);
                PatrolResultSummaryVo summary = patrolResultService.summary(query);
                if (summary == null) {
                    return;
                }
                List<PatrolResultEntity> data = patrolResultService.list(
                        Wrappers.<PatrolResultEntity>lambdaQuery().eq(PatrolResultEntity::getTicketsId, ticketsId)
                                .orderByAsc(PatrolResultEntity::getRoomName)
                                .orderByAsc(PatrolResultEntity::getDevName)
                );
                List<PatrolResultExcelDto> list = new ArrayList<>();
                //属性替换
                for (int i = 0; i < data.size(); i++) {
                    PatrolResultEntity patrolResult = data.get(i);
                    PatrolResultExcelDto excelDto = BeanUtil.copyProperties(patrolResult, PatrolResultExcelDto.class);
                    excelDto.setIndex(i + 1);//序号
                    String url = StrUtil.isBlank(patrolResult.getVisionMediaUrl()) ? patrolResult.getMediaUrl() : patrolResult.getVisionMediaUrl();
                    if (StrUtil.isNotBlank(url)) {
                        //服务端的链接带了sys-file
                        url = url.replace("/sys-file/", "");
                    }
                    excelDto.setPath(fileProperties.getLocal().getBasePath() + File.separator + url);//文件采用相对路径，所以要加前缀
                    list.add(excelDto);
                }
                //动态表头
                List<List<String>> headTitles = ExcelUtil.head(summary);
                //写文件
                EasyExcel.write(patrolExcelFilePath, PatrolResultExcelDto.class)
                        // 这里放入动态头
                        .head(headTitles)
                        .sheet("Sheet1")
                        .registerWriteHandler(new LongestMatchColumnWidthStyleStrategy())
                        .registerWriteHandler(new MultiColumnMergeHandler(data.size(), Arrays.asList(1, 2)))//合并
                        // 当然这里数据也可以用 List<List<String>> 去传入
                        .doWrite(list);

            }
        }
    }

    @EventListener
    @Async("serverListenerThreadPool")
    public void handleQuery(MessageDto messageDto) throws IOException {
        ProtocolEnum protocolEnum = ProtocolEnum.CMD_PROTOCOL_MAP.get(messageDto.getCmd());
        //log.info("[netty listener]" + messageDto.getSn());
        String clientId = messageDto.getClientId();
        if (messageDto.getSn() > 0) {
            Lock lock = LockUtil.ROBOT_CMD_TICKETS_LOCKS.get(clientId, (short) messageDto.getSn());
            if (lock != null) {
                RobotReplyUtil.COMMON_REPLY_TABLE.put(clientId, (short) messageDto.getSn(),
                        JSONObject.parseObject(new String(messageDto.getData(), StandardCharsets.UTF_8), R.class));
                try {
                    lock.notifyAll();
                } catch (Exception e) {

                }
            }
        }

    }


    @EventListener
    @Async("serverListenerThreadPool")
    public void handlerQuery(RobotCmdNavDto robotCmdNavDto) {


    }


    //处理分块文件
    @EventListener
    @Async("serverFileThreadPool")
    public void handlerFile(FileChunkDto dto) {


    }


    @EventListener
    @Async("serverListenerThreadPool")
    public void handlePatrolInfo(PatrolConfigInfo dto) {
        if (dto == null || StrUtil.isBlank(dto.getRobotSn())) {
            return;
        }
        RedisUtils.set("patrol:dev:" + dto.getRobotSn(), JSONObject.toJSONString(dto.getDevs()));
        RedisUtils.set("patrol:checkTypes:" + dto.getRobotSn(), JSONObject.toJSONString(dto.getCheckTypes()));
        RedisUtils.set("patrol:objects:" + dto.getRobotSn(), JSONObject.toJSONString(dto.getObjects()));
        RedisUtils.set("patrol:posPoseGroups:" + dto.getRobotSn(), JSONObject.toJSONString(dto.getPosPoseGroups()));
        RedisUtils.set("patrol:items:" + dto.getRobotSn(), JSONObject.toJSONString(dto.getItems()));
        log.info("保存{}巡检配置", dto.getRobotSn());
    }


    @EventListener
    @Async("serverListenerThreadPool")
    public void handlerWarning(RobotWarningDto robotWarningDto) {
        log.info("Robot Warning" + JSONObject.toJSONString(robotWarningDto));
        RobotWarningRecordEntity robotWarningRecord = new RobotWarningRecordEntity();
        robotWarningRecord.setFlag("0");
        robotWarningRecord.setContent(robotWarningDto.getContent());
        robotWarningRecord.setServiceId(robotWarningDto.getRobotSn());
        robotWarningRecord.setTime(LocalDateTime.now());
        robotWarningRecord.setTitle(robotWarningDto.getWarningObj());
        robotWarningRecord.setType("0");
        if (robotWarningRecord.getTitle().contains("检测")) {
            robotWarningRecord.setType("2");
        }
        robotWarningRecord.setLevel(robotWarningDto.getLevel());
        robotWarningRecord.setRemark(robotWarningDto.getRemark());
        String paths = robotWarningDto.getPaths();
        if (StrUtil.isNotBlank(paths)) {
            String[] split = paths.split(",");
            for (String s : split) {
                String extName = FileUtil.extName(s);
                String filePath = fileProperties.getLocal().getBasePath() + File.separator + "warning" + File.separator + robotWarningDto.getRobotSn() + File.separator + DateUtil.format(new Date(), "yyyy-MM-dd") + File.separator + DateUtil.format(new Date(), "HHmmssSSS") + "." + extName;
                String relativePath = normalizePath(filePath).replace(normalizePath(fileProperties.getLocal().getBasePath()), "/sys-file");
                try {
                    URL url = URLUtil.url(s);
                    FileUtil.writeFromStream(URLUtil.getStream(url), filePath);
                    paths = paths.replace(s, relativePath);
                } catch (IORuntimeException e) {
                    log.error("下载报警媒体失败,{}", e.getMessage());
                }
            }
        }
        robotWarningRecord.setFiles(paths);
        robotWarningRecordService.save(robotWarningRecord);
    }

    public static void broadcastMessageByTopic(String topic) {
        //限流
        RateLimiter rateLimiter = topicRateLimiterMap.computeIfAbsent(topic, k -> RateLimiter.create(1));
        if (rateLimiter.tryAcquire()) {
            Map<String, String> clientMessages = topicClientMessages.getOrDefault(topic, Collections.emptyMap());
            // 创建广播的消息内容，包含所有去重后的消息
            String broadcastMessage = createBroadcastMessage(clientMessages);
            // 广播消息给所有订阅的会话
            SubscriptionEndpoint.broadcast(topic, broadcastMessage);
        }
    }

    private static String createBroadcastMessage(Map<String, String> clientMessages) {
        try {
            // 将去重后的消息转换成JSONArray
            ArrayNode messagesArray = objectMapper.createArrayNode();
            //按key的顺序去添加
            clientMessages.entrySet().stream().sorted(Map.Entry.comparingByKey()).forEach(entry -> {
                try {
                    messagesArray.add(objectMapper.readTree(entry.getValue()));
                } catch (IOException e) {
                    log.error("[ws][publish]Failed to parse message: {}", e.getMessage());
                }
            });
            //转成不带换行符的json字符串
            return objectMapper.writeValueAsString(messagesArray);
        } catch (Exception e) {
            log.error("[ws][publish]Failed to create broadcast message: {}", e.getMessage());
            return "{\"error\":\"Failed to create broadcast message\"}";
        }
    }

    /**
     * 移除所有topic中指定执行端相关消息
     *
     * @param clientId
     */
    public static void removeClientMsgFromAllTopic(String clientId) {
        if (StringUtils.hasText(clientId)) {
            List<Map<String, String>> listMap = topicClientMessages.values().stream().toList();
            for (Map<String, String> map : listMap) {
                if (map.containsKey(clientId)) {
                    map.remove(clientId);
                }
            }
        }
    }


    // 工具方法：将路径统一转为 "/" 分隔
    private String normalizePath(String path) {
        return path.replace("\\", "/");
    }
}
